package com.tencent.cloud.iov.flow.loader.impl;

import android.support.annotation.NonNull;

import com.tencent.cloud.iov.flow.loader.ILoader;
import com.tencent.cloud.iov.util.rx.RxUtils;
import com.tencent.cloud.iov.util.rx.SimpleObserver;

import java.util.HashMap;
import java.util.Map;

import io.reactivex.Notification;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.subjects.BehaviorSubject;

/**
 * 模型抽象类
 *
 * 实现数据订阅及取消订阅逻辑，管理数据订阅状态
 *
 * 数据加载业务与数据订阅业务是分离的：使用中介者 {@link #mBehaviorSubject} 进行实现
 *
 * 订阅者订阅数据 {@link #register(Observer)} 后，仅能接收到订阅之后发射的数据
 * 取消数据订阅时 {@link #unregister(Observer)}，仅会结束数据通知业务，数据加载业务仍会执行
 * 如果希望结束数据加载业务，需要通过 {@link #close()} 关闭数据模型
 *
 * @author showxiao
 * @date 2018/3/27
 */

public abstract class BaseFlowLoader<T> implements ILoader<Observer<T>> {

    /**
     * 中介者：连接数据加载业务与数据订阅业务
     *
     * 为了避免 Subject 传递错误（onError）后，订阅关系不再生效，所以将数据或者错误通过 Notification 进行包装
     */
    private BehaviorSubject<Notification<T>> mBehaviorSubject = BehaviorSubject.create();

    /** 数据订阅状态管理集合 */
    private Map<Observer<T>, Disposable> mRegisterDisposables = new HashMap<>();

    /** （接收）数据观察者 */
    private Observer<T> mDataObserver;

    /** 数据加载业务管理集合 */
    private Disposable mDataDisposable;

    @Override
    public void register(Observer<T> observer) {
        if (!mRegisterDisposables.containsKey(observer)) {
            mBehaviorSubject
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new OnNextErrorObserver(observer));
        }
    }

    @Override
    public void loadData() {
        getDataObservable().subscribe(getDataObserver());
    }

    /**
     * 请求数据
     *
     * @return
     */
    @NonNull
    protected abstract Observable<T> getDataObservable();

    @NonNull
    protected final Observer<T> getDataObserver() {
        if (mDataObserver == null) {
            mDataObserver = new OnErrorNextObserver();
        }
        return mDataObserver;
    }

    @Override
    public void unregister(final Observer<T> observer) {
        if (mRegisterDisposables.containsKey(observer)) {
            Disposable disposable = mRegisterDisposables.remove(observer);
            disposable.dispose();
        }
    }

    @Override
    public void close() {
        // 多观察者情况下不关闭数据请求
        if (hasObservers()) {
            return;
        }

        // 取消所有订阅状态
        for (Disposable disposable : mRegisterDisposables.values()) {
            RxUtils.dispose(disposable);
        }
        mRegisterDisposables.clear();
        mRegisterDisposables = null;

        // 取消所有数据加载业务
        RxUtils.dispose(mDataDisposable);
        mDataDisposable = null;
    }

    protected final boolean hasObservers() {
        return mRegisterDisposables != null && !mRegisterDisposables.isEmpty();
    }

    /**
     *  （发送）数据观察者
     *
     *  根据包装后的数据（Notification）类型发送数据或者错误
     */
    private class OnNextErrorObserver extends SimpleObserver<Notification<T>> {

        private Observer<T> mActual;

        OnNextErrorObserver(Observer<T> actual) {
            mActual = actual;
        }

        @Override
        public void onSubscribe(Disposable d) {
            super.onSubscribe(d);
            mRegisterDisposables.put(mActual, d);
        }

        @Override
        public void onNext(Notification<T> notification) {
            super.onNext(notification);

            if (notification.isOnError()) {
                mActual.onError(notification.getError());
            } else {
                mActual.onNext(notification.getValue());
            }
        }
    }

    /**
     * （接收）数据观察者
     *
     * 将数据或者错误包装后（Notification）进行发送
     * 避免 Subject 传递错误（onError）后，订阅关系不再生效
     */
    private class OnErrorNextObserver extends SimpleObserver<T> {

        @Override
        public void onSubscribe(Disposable d) {
            super.onSubscribe(d);
            RxUtils.dispose(mDataDisposable);
            mDataDisposable = d;
        }

        @Override
        public void onNext(T t) {
            super.onNext(t);
            mBehaviorSubject.onNext(Notification.createOnNext(t));
        }

        @Override
        public void onError(Throwable e) {
            super.onError(e);
            mBehaviorSubject.onNext(Notification.<T>createOnError(e));
        }

    }
}
